package com.swak.job.queue.guava;

import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Function;

import com.google.common.eventbus.CustomEventBus;
import com.google.common.eventbus.EventBus;
import com.google.common.eventbus.Subscribe;
import com.swak.OS;
import com.swak.job.JobTask;
import com.swak.job.queue.QueueService;
import com.swak.utils.Maps;
import com.swak.utils.Sets;

import lombok.Getter;
import lombok.Setter;
import lombok.experimental.Accessors;

/**
 * GuavaQueueService.
 * 
 * @author 618lf
 */
public class GuavaQueueService implements QueueService {

	Map<String, MessageConsumer> queueConsumers = Maps.newConcurrentMap();

	private final EventBus eventBus;
	private final ExecutorService executor;

	public GuavaQueueService() {
		executor = Executors.newFixedThreadPool(OS.eventLoops(4),
				new EventLoopFactory(true, "JOB.eventbus", new AtomicLong()));
		eventBus = new CustomEventBus("JOB.eventbus", executor);
		eventBus.register(this);
	}

	/**
	 * 处理事件
	 * 
	 * @param event
	 */
	@Subscribe
	public void handle(JobTaskEvent event) {
		String queue = event.getQueue();
		MessageConsumer consumer = queueConsumers.get(queue);
		if (consumer != null) {
			consumer.handle(event);
		}
	}

	@Override
	public void createQueue(String queue) {
	}

	@Override
	public void publishQueue(String queue, JobTask task) {
		executor.execute(() -> {
			this.eventBus.post(JobTaskEvent.of(queue, task));
		});
	}

	@Override
	public CompletableFuture<Void> publishQueueAsync(String queue, JobTask task) {
		CompletableFuture<Void> future = new CompletableFuture<>();
		executor.execute(() -> {
			this.eventBus.post(JobTaskEvent.of(queue, task));
			future.complete(null);
		});
		return future;
	}

	@Override
	public void consumeQueue(String queue, Function<JobTask, Object> func) {
		MessageConsumer messageConsumer = queueConsumers.computeIfAbsent(queue, (key) -> {
			return new MessageConsumer(queue);
		});
		messageConsumer.addHandler(func);
	}

	@Override
	public void unconsumeQueue(String queue, Function<JobTask, Object> func) {
		MessageConsumer messageConsumer = queueConsumers.computeIfAbsent(queue, (key) -> {
			return new MessageConsumer(queue);
		});
		messageConsumer.delHandler(func);
	}

	@Getter
	@Setter
	@Accessors(chain = true)
	public static class MessageConsumer {

		private String queue;
		private Set<Function<JobTask, Object>> handlers;

		public MessageConsumer(String queue) {
			this.queue = queue;
			this.handlers = Sets.newConcurrentSet();
		}

		public MessageConsumer addHandler(Function<JobTask, Object> handler) {
			this.handlers.add(handler);
			return this;
		}

		public MessageConsumer delHandler(Function<JobTask, Object> handler) {
			this.handlers.remove(handler);
			return this;
		}

		public void handle(JobTaskEvent event) {
			for (Function<JobTask, Object> handler : handlers) {
				try {
					handler.apply(event.getTask());
				} catch (Exception e) {
				}
			}
		}
	}

	@Getter
	@Setter
	@Accessors(chain = true)
	public static class JobTaskEvent {
		private String queue;
		private JobTask task;

		public static JobTaskEvent of(String queue, JobTask task) {
			return new JobTaskEvent().setQueue(queue).setTask(task);
		}
	}

	public static class EventLoopFactory implements ThreadFactory {

		final boolean daemon;
		final AtomicLong counter;
		final String prefix;

		public EventLoopFactory(boolean daemon, String prefix, AtomicLong counter) {
			this.daemon = daemon;
			this.counter = counter;
			this.prefix = prefix;
		}

		@Override
		public Thread newThread(Runnable r) {
			Thread t = new Thread(r);
			t.setDaemon(daemon);
			t.setName(prefix + "-" + counter.incrementAndGet());
			return t;
		}
	}
}